package elasticsearch;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.nlpcn.es4sql.exception.SqlParseException;
import org.nlpcn.es4sql.jdbc.ObjectResult;
import org.nlpcn.es4sql.jdbc.ObjectResultsExtractException;
import org.nlpcn.es4sql.query.ESActionFactory;
import org.nlpcn.es4sql.query.QueryAction;
import org.nlpcn.es4sql.query.SqlElasticRequestBuilder;

import java.io.IOException;
import java.sql.SQLFeatureNotSupportedException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

@Slf4j
public class EsClient {

    private RestHighLevelClient restHighLevelClient;

    public SearchRequest parse(String dsl) throws SQLFeatureNotSupportedException, SqlParseException {
        QueryAction queryAction = ESActionFactory.create(null, dsl);
        SqlElasticRequestBuilder requestBuilder = queryAction.explain();
        return (SearchRequest) requestBuilder.request();
    }

    public SearchHit[] queryByDsl(DslBody dslBody) {
        SearchResponse searchResponse;
        try {
            SearchRequest searchRequest = parse(dslBody.getDsl());
            searchRequest.source().trackTotalHits(true);
            // 不启用stored_fields
            if (ObjectUtils.isNotEmpty(searchRequest.source().storedFields())
                    && CollectionUtils.isNotEmpty(searchRequest.source().storedFields().fieldNames())) {
                searchRequest.source().storedFields().fieldNames().clear();
            }
            if (StringUtils.isNotBlank(dslBody.getShardingKeys())) {
                searchRequest.routing(dslBody.getShardingKeys().split(","));
            }
//            if (localPreferEnabled) {
//                searchRequest.preference(LOCAL_READ_PREFERENCE);
//            } else {
//                if (StringUtils.isNotBlank(dslBody.getUserId())) {
//                    searchRequest.preference(dslBody.getUserId());
//                }
//            }
            if (dslBody.isScroll()) {
                searchRequest.scroll(TimeValue.timeValueMinutes(1));
            }
            if (ArrayUtils.isNotEmpty(dslBody.getSearchAfter())) {
                searchRequest.source().searchAfter(dslBody.getSearchAfter());
            }
            searchRequest.searchType(SearchType.DEFAULT);
            String[] indices = searchRequest.indices();
            searchRequest.indices(Arrays.stream(indices).toArray(String[]::new));

            if (!dslBody.getOrderMap().isEmpty() && searchRequest.source().sorts() != null) {
                for (SortBuilder sort : searchRequest.source().sorts()) {
                    if (sort instanceof FieldSortBuilder) {
                        FieldSortBuilder fieldSortBuilder = (FieldSortBuilder) sort;
                        fieldSortBuilder.unmappedType(dslBody.getOrderMap().get(fieldSortBuilder.getFieldName()));
                    }
                }
            }
            searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        } catch (SQLFeatureNotSupportedException | SqlParseException e) {
            log.error("es params parse error", e);
            throw new IllegalArgumentException("requestparams.invald");
        } catch (IOException e) {
            log.error("es query error", e);
            throw new IllegalStateException("elasticsearch server error", e);
        }

        return searchResponse.getHits().getHits();
    }


//    private ObjectResult extractResults(Object queryResult, boolean flat) throws ObjectResultsExtractException {
//        if (queryResult instanceof SearchHits) {
//            SearchHit[] hits = ((SearchHits) queryResult).getHits();
//            List<Map<String, Object>> docsAsMap = new ArrayList<>();
//            List<String> headers = createHeadersAndFillDocsMap(flat, hits, null, docsAsMap);
//            List<List<Object>> lines = createLinesFromDocs(flat, docsAsMap, headers);
//            return new ObjectResult(headers, lines);
//        }
//        if (queryResult instanceof Aggregations) {
//            List<String> headers = new ArrayList<>();
//            List<List<Object>> lines = new ArrayList<>();
//            lines.add(new ArrayList<Object>());
//            handleAggregations((Aggregations) queryResult, headers, lines);
//
//            // remove empty line。
//            if (lines.get(0).size() == 0) {
//                lines.remove(0);
//            }
//            //todo: need to handle more options for aggregations:
//            //Aggregations that inhrit from base
//            //ScriptedMetric
//
//            return new ObjectResult(headers, lines);
//
//        }
//        if (queryResult instanceof SearchResponse) {
//            SearchHit[] hits = ((SearchResponse) queryResult).getHits().getHits();
//            List<Map<String, Object>> docsAsMap = new ArrayList<>();
//            List<String> headers = createHeadersAndFillDocsMap(flat, hits, ((SearchResponse) queryResult).getScrollId(), docsAsMap);
//            List<List<Object>> lines = createLinesFromDocs(flat, docsAsMap, headers);
//            return new ObjectResult(headers, lines);
//        }
//        return null;
//    }

    private List<List<Object>> createLinesFromDocs(boolean flat, List<Map<String, Object>> docsAsMap, List<String> headers) {
        List<List<Object>> objectLines = new ArrayList<>();
        for (Map<String, Object> doc : docsAsMap) {
            List<Object> lines = new ArrayList<>();
            for (String header : headers) {
                lines.add(findFieldValue(header, doc, flat));
            }
            objectLines.add(lines);
        }
        return objectLines;
    }


    private Object findFieldValue(String header, Map<String, Object> doc, boolean flat) {
        if (flat && header.contains(".")) {
            String[] split = header.split("\\.");
            Object innerDoc = doc;
            for (String innerField : split) {
                if (!(innerDoc instanceof Map)) {
                    return null;
                }
                innerDoc = ((Map<String, Object>) innerDoc).get(innerField);
                if (innerDoc == null) {
                    return null;
                }

            }
            return innerDoc;
        } else {
            if (doc.containsKey(header)) {
                return doc.get(header);
            }
        }
        return null;
    }

}
